/*
    主题发布订阅业务层服务端：能够支持主题的创建/删除/订阅/取消订阅，以及对消息的发布进行转发的能力
    1、发布订阅管理类：
        设计思想：
            1、提供一个主题发布/订阅的回调，被注册给Dispatcher模块：当接收到客户端的发布主题/订阅主题的请求时调用
            2、提供一个清理订阅者信息的回调，被注册给Dispatcher模块：当连接断开时清理订阅者信息时调用
            3、需要维护<主题，主题描述>关系，当有发布了主题消息的时，能够对订阅了该主题的客户端进行消息的转发
            4、需要维护<连接，订阅者>关系，当有连接断开的时候，能够找到订阅者，根据他订阅的主题，在订阅者列表中删除其订阅的信息
            注意：为了方便发布订阅管理类处理请求，以及维护主题与订阅者之间的关系，将这部分抽离成主题管理类

    2、主题管理类：根据主题的创建、删除、订阅、取消订阅来对主题与订阅者的管理，其内部包含两个描述类：主题类和订阅者
        2.1、根据主题的创建、删除、订阅、取消订阅，提供对应的接口
        2.2、当连接断开的时候提供一个接口，能够根据连接，找到订阅者，清理订阅者的信息
        2.3、主题类：应该具有主题名、订阅者列表，以及操作订阅者列表的接口
        2.4、订阅者类：应该具有连接、主题列表，以及操作主题列表的接口
        2.5、<主题名，主题类>，<连接，订阅者>两张关系表
*/
#pragma once
#include <unordered_set>
#include <unordered_map>
#include <set>
#include "../common/message.hpp"

namespace ns_jsonrpc
{
    namespace ns_server
    {
        // 主题管理
        class TopicManager
        {
        public:
            using TopicManagerPtr = std::shared_ptr<TopicManager>;

        private:
            // 订阅者
            struct Subscriber
            {
                using SubscriberPtr = std::shared_ptr<Subscriber>;
                BaseConnection::BaseConnectionPtr conn;
                std::unordered_set<std::string> topics;
                std::mutex mtx;

                Subscriber(const BaseConnection::BaseConnectionPtr &conn) : conn(conn)
                {
                }

                // 主题订阅的时候，将主题加入到当前订阅者的订阅列表中
                void appendTopic(const std::string &topic)
                {
                    std::unique_lock<std::mutex> lock(mtx);
                    topics.insert(topic);
                }

                // 取消订阅的时候，将主题从当前订阅者的订阅列表中删除
                void removeTopic(const std::string &topic)
                {
                    std::unique_lock<std::mutex> lock(mtx);
                    topics.erase(topic);
                }
            };

            // 主题
            struct Topic
            {
                using TopicPtr = std::shared_ptr<Topic>;
                std::string topic_name;
                std::unordered_set<Subscriber::SubscriberPtr> subscribers;
                std::mutex mtx;

                Topic(const std::string &topic_name) : topic_name(topic_name)
                {
                }

                // 主题订阅的时候添加订阅者
                void appendSubscriber(const Subscriber::SubscriberPtr &subscriber)
                {
                    std::unique_lock<std::mutex> lock(mtx);
                    subscribers.insert(subscriber);
                }

                // 取消订阅的时候删除订阅者/订阅者连接断开的时候删除订阅者
                void removeSubscriber(const Subscriber::SubscriberPtr &subscriber)
                {
                    std::unique_lock<std::mutex> lock(mtx);
                    subscribers.erase(subscriber);
                }

                // 消息发布的时候，进行消息的转发
                void pushMessage(const BaseMessage::BaseMessagePtr &msg)
                {
                    std::unique_lock<std::mutex> lock(mtx);
                    for (auto &subscriber : subscribers)
                        subscriber->conn->send(msg);
                }
            };

        public:
            // 主题创建（采用强断言的方式处理，没有则创建，有则忽略）
            void topicCreate(const std::string &topicName, const BaseConnection::BaseConnectionPtr &conn)
            {
                std::unique_lock<std::mutex> lock(mtx);
                _topics.insert({topicName, std::make_shared<Topic>(topicName)});
            }

            // 主题删除（采用强断言的方式处理，有则删除，没有则忽略）
            void topicRemove(const std::string &topicName, const BaseConnection::BaseConnectionPtr &conn)
            {
                // 删除主题 & 订阅了该主题的所有订阅者的订阅主题信息
                std::vector<Subscriber::SubscriberPtr> sbers;
                {
                    std::unique_lock<std::mutex> lock(mtx);
                    auto it = _topics.find(topicName);
                    if (it == _topics.end())
                        return;
                    for (auto &subscriber : it->second->subscribers)
                        sbers.push_back(subscriber);
                    _topics.erase(it);
                }
                for (auto &subscriber : sbers)
                    subscriber->removeTopic(topicName);
            }

            // 主题订阅
            bool topicSubscribe(const std::string topicName, const BaseConnection::BaseConnectionPtr &conn)
            {
                // 在主题中添加订阅者信息 & 在订阅者中添加主题信息
                Topic::TopicPtr topic;
                Subscriber::SubscriberPtr subscriber;
                {
                    // 先找到主题，如果没有则报错
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto itTopic = _topics.find(topicName);
                    if (itTopic == _topics.end())
                        return false;
                    topic = itTopic->second;

                    // 再找到订阅者，如果没有则创建
                    auto itSubscriber = _subscribers.find(conn);
                    if (itSubscriber == _subscribers.end())
                        subscriber = std::make_shared<Subscriber>(conn);
                    else
                        subscriber = itSubscriber->second;
                }
                topic->appendSubscriber(subscriber);
                subscriber->appendTopic(topicName);
                return true;
            }

            // 主题取消订阅
            void topicCancel(const std::string &topicName, const BaseConnection::BaseConnectionPtr &conn)
            {
                // 删除主题中的订阅者信息 & 删除订阅者中的主题信息
                Topic::TopicPtr topic;
                Subscriber::SubscriberPtr subscriber;
                {
                    // 找到主题
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto itTopic = _topics.find(topicName);
                    if (itTopic == _topics.end())
                        return;
                    topic = itTopic->second;

                    // 再找到订阅者
                    auto itSubscriber = _subscribers.find(conn);
                    if (itSubscriber != _subscribers.end())
                        subscriber = itSubscriber->second;
                }
                topic->removeSubscriber(subscriber);
                if (subscriber != nullptr)
                    subscriber->removeTopic(topicName);
            }

            // 主题消息发布
            bool topicPublish(const std::string &topicName, const BaseMessage::BaseMessagePtr &msg)
            {
                // 找到主题，根据内部的订阅者列表进行消息的转发
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _topics.find(topicName);
                if (it == _topics.end())
                    return false;
                it->second->pushMessage(msg);
                return true;
            }

            // 当订阅者连接断开的时候，调用
            void onConnshutdown(const BaseConnection::BaseConnectionPtr &conn)
            {
                // 找到连接者
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _subscribers.find(conn);
                if (it == _subscribers.end())
                    return;
                
                // 删除主题对象中的订阅者信息
                for(auto &topicName : it->second->topics)
                {
                    auto it2 = _topics.find(topicName);
                    if (it2 != _topics.end())
                        it2->second->removeSubscriber(it->second);
                }
            }

        private:
            std::unordered_map<std::string, Topic::TopicPtr> _topics;                                      // <主题名，主题描述>
            std::unordered_map<BaseConnection::BaseConnectionPtr, Subscriber::SubscriberPtr> _subscribers; // <连接，订阅者>
            std::mutex _mutex;
        };

        // 主题发布订阅管理
        class PSManager
        {
        public:
            using PSManagerPtr = std::shared_ptr<PSManager>;

        public:
            PSManager() : _topic_manager(std::make_shared<TopicManager>())
            {
            }

            // 接收到主题请求的时候，调用的回调函数，被注册给Dispatcher模块
            void onTopicRequest(const BaseConnection::BaseConnectionPtr &conn, const TopicRequest::TopicRequestPtr &msg)
            {
                // body检查
                if (!msg->check())
                {
                    LOG_ERROR("topic request failed, invalid message\n");
                    return;
                }
                // 根据主题的操作类型，分别处理
                TopicOpType optype = msg->opType();
                bool ret = true;
                switch (optype)
                {
                // 主题创建
                case TopicOpType::TOPIC_CREATE:
                    _topic_manager->topicCreate(msg->key(), conn);
                    break;

                // 主题删除
                case TopicOpType::TOPIC_REMOVE:
                    _topic_manager->topicRemove(msg->key(), conn);
                    break;

                // 主题订阅
                case TopicOpType::TOPIC_SUBSCRIBE:
                    ret = _topic_manager->topicSubscribe(msg->key(), conn);
                    break;

                // 主题取消订阅
                case TopicOpType::TOPIC_CANCEL:
                    _topic_manager->topicCancel(msg->key(), conn);
                    break;

                // 主题消息发布
                case TopicOpType::TOPIC_PUBLISH:
                    ret = _topic_manager->topicPublish(msg->key(), msg);
                    break;

                default:
                    responseError(conn, msg, RCode::RCODE_INVALID_OPTYPE);
                    break;
                }
                if (!ret)
                    return responseError(conn, msg, RCode::RCODE_NOT_FOUND_TOPIC);
                response(conn, msg);
            }

            // 当连接关闭的时候的回调，用来清理订阅者的相关信息
            void onConnShutdown(const BaseConnection::BaseConnectionPtr &conn)
            {
                _topic_manager->onConnshutdown(conn);
            }

        private:
            void response(const BaseConnection::BaseConnectionPtr &conn, const TopicRequest::TopicRequestPtr &msg)
            {
                TopicResponse::TopicResponsePtr rsp = MessageFactory::create<TopicResponse>();
                rsp->setId(msg->id());
                rsp->setMType(MType::RSP_TOPIC);
                rsp->setRCode(RCode::RCODE_OK);
                conn->send(rsp);
            }

            void responseError(const BaseConnection::BaseConnectionPtr &conn, const TopicRequest::TopicRequestPtr &msg, RCode rcode)
            {
                TopicResponse::TopicResponsePtr rsp = MessageFactory::create<TopicResponse>();
                rsp->setId(msg->id());
                rsp->setMType(MType::RSP_TOPIC);
                rsp->setRCode(rcode);
                conn->send(rsp);
            }

        private:
            TopicManager::TopicManagerPtr _topic_manager;
        };
    }
}